package com.lagou.no4

import org.apache.flink.api.java.tuple.Tuple
import org.apache.flink.streaming.api.TimeCharacteristic
import org.apache.flink.streaming.api.scala.{DataStream, StreamExecutionEnvironment, WindowedStream}
import org.apache.flink.api.scala._
import org.apache.flink.streaming.api.functions.timestamps.BoundedOutOfOrdernessTimestampExtractor
import org.apache.flink.streaming.api.scala.function.WindowFunction
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows
import org.apache.flink.streaming.api.windowing.time.Time
import org.apache.flink.streaming.api.windowing.windows.TimeWindow
import org.apache.flink.util.Collector
case class WaterMark(id:Int,time:Long)
object WaterMark {
    def main(args: Array[String]): Unit = {
        val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment
        //指定事件时间
        env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime)
        //水印生成周期
        env.getConfig.setAutoWatermarkInterval(200)
        //设置并行度为1
        env.setParallelism(1)
        //连接socket数据流
        val dataDS: DataStream[String] = env.socketTextStream("node01", 9999)
        //将数据流封装成样例类
        val wcDs: DataStream[WaterMark] = dataDS.map { line =>
            val arr: Array[String] = line.split(",")
            WaterMark(arr(0).trim.toInt, arr(1).trim.toLong)
        }
        //注册水印 ，设置允许迟到为2秒
        val waterMarkDS = wcDs.assignTimestampsAndWatermarks(new BoundedOutOfOrdernessTimestampExtractor[WaterMark](Time.seconds(2)) {
            //返回事件时间字段
            override def extractTimestamp(t: WaterMark): Long = {
                t.time
            }
        })
        //设置窗口为5秒
        val windowDS: WindowedStream[WaterMark, Tuple, TimeWindow] = waterMarkDS.keyBy(0).window(TumblingEventTimeWindows.of(Time.seconds(5)))
        //apply方法。输出样例类对象
        val res: DataStream[WaterMark] = windowDS.apply(new WindowFunction[WaterMark, WaterMark, Tuple, TimeWindow] {
            override def apply(key: Tuple, window: TimeWindow, input: Iterable[WaterMark], out: Collector[WaterMark]): Unit = {
                //打印开始时间和结束时间
                println(s"window开始时间${window.getStart},window结束时间${window.getEnd}")
                //输出样例类
                input.foreach(x => out.collect(x))
            }
        })
        //打印
        res.print()

        env.execute()

    }
}
